package com.xframe.xdal.core.listener.mysql;

import com.github.shyiko.mysql.binlog.event.*;
import com.github.shyiko.mysql.binlog.network.ServerException;
import com.xframe.xdal.core.DataBaseAccess;
import com.xframe.xdal.core.listener.DataBaseListener;
import com.xframe.xdal.core.listener.model.ConnectorConstant;
import com.xframe.xdal.core.manage.DBTableInfo;
import com.xframe.xdal.core.util.StringUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
// SET GLOBAL binlog_format = 'ROW';
public class MySqlListener extends DataBaseListener {

    private final Logger logger = LoggerFactory.getLogger(getClass());

    private static final String BINLOG_FILENAME = "fileName";
    private static final String BINLOG_POSITION = "position";
    private static final int RETRY_TIMES = 10;

    private BinaryLogRemoteClient client;
    private String fileName = "";
    private Long pos = 0L;

    private final Lock connectLock = new ReentrantLock();
    private volatile boolean connected;

    public MySqlListener(DataBaseAccess dataBaseAccess){
        super(dataBaseAccess);

    }

    @Override
    public void start(){
        try {
            connectLock.lock();
            if (connected) {
                logger.error("MySqlListener is already started");
                return;
            }
            run();
            connected = true;
        } catch (Exception exception) {
            exception.printStackTrace();
        }finally {
            connectLock.unlock();
        }
    }



    private void run() throws Exception {

        tables = dataBaseAccess.getDataSource().getDbManage().getTableInfoMap();
        client = new BinaryLogRemoteClient(dataBaseAccess.getDataSource().getHost(),
                dataBaseAccess.getDataSource().getPort(), dataBaseAccess.getDataSource().getUsername(),
                dataBaseAccess.getDataSource().getPassword(),dataBaseAccess.getDataSource().getDatabaseName());
        client.setBinlogFilename(snapshot.get(BINLOG_FILENAME));
        client.setBinlogPosition(pos);

        client.registerEventListener(new MysqlEventListener());
        client.registerLifecycleListener(new MysqlLifecycleListener());

        client.connect();
    }

    private void reStart() {
        for (int i = 1; i <= RETRY_TIMES; i++) {
            try {
                if (null != client) {
                    client.disconnect();
                }
                run();

                //errorEvent(new ListenerException(String.format("重启成功, %s", client.getWorkerThreadName())));
                logger.error("第{}次重启成功, ThreadName:{} ", i, client.getWorkerThreadName());
                break;
            } catch (Exception e) {
                logger.error("第{}次重启异常, ThreadName:{}, {}", i, client.getWorkerThreadName(), e.getMessage());
                // 无法连接，关闭任务
                if (i == RETRY_TIMES) {
                    logger.error("重启异常, {}, {}", client.getWorkerThreadName(), e.getMessage());
                    //interruptException(new ListenerException(String.format("重启异常, %s, %s", client.getWorkerThreadName(), e.getMessage())));
                }
            }
            try {
                TimeUnit.SECONDS.sleep(i * 2);
            } catch (InterruptedException e) {
                logger.error(e.getMessage());
            }
        }
    }

    private void refresh(EventHeader header) {
        EventHeaderV4 eventHeaderV4 = (EventHeaderV4) header;
        refresh(null, eventHeaderV4.getNextPosition());
    }

    private void refresh(String binlogFilename, long nextPosition) {
        if (StringUtil.isNotBlank(binlogFilename)) {
            client.setBinlogFilename(binlogFilename);
            snapshot.put(BINLOG_FILENAME, binlogFilename);
        }
        if (0 < nextPosition) {
            client.setBinlogPosition(nextPosition);
            snapshot.put(BINLOG_POSITION, String.valueOf(nextPosition));
        }
    }

    @Override
    public void close() {
        try {
            connectLock.lock();
            connected = false;
            if (null != client) {
                client.disconnect();
            }
        } catch (Exception e) {
            logger.error("关闭失败:{}", e.getMessage());
        } finally {
            connectLock.unlock();
        }
    }

    final class MysqlLifecycleListener implements BinaryLogRemoteClient.LifecycleListener {

        @Override
        public void onConnect(BinaryLogRemoteClient client) {
            // 记录日志的位置 和名字
            fileName = client.getBinlogFilename();
            pos = client.getBinlogPosition();
            refresh(client.getBinlogFilename(), client.getBinlogPosition());
            logger.info("fileName = {},pos = {}",fileName,pos);
        }

        @Override
        public void onCommunicationFailure(BinaryLogRemoteClient client, Exception e) {
            // 通信失败事件
            // 可以加入重连的代码
            if (!client.isConnected()) {
                return;
            }
            logger.error(e.getMessage());
            if (e instanceof ServerException) {
                ServerException serverException = (ServerException) e;
                if (serverException.getErrorCode() == 1236) {
                    close();
                    String log = String.format("线程[%s]执行异常。由于MySQL配置了过期binlog文件自动删除机制，已无法找到原binlog文件%s。建议先保存驱动（加载最新的binlog文件），再启动驱动。",
                            client.getWorkerThreadName(),
                            client.getBinlogFilename());
                    //interruptException(new ListenerException(log));
                    return;
                }
            }

            reStart();
        }

        @Override
        public void onEventDeserializationFailure(BinaryLogRemoteClient client, Exception ex) {

        }

        @Override
        public void onDisconnect(BinaryLogRemoteClient client) {

        }
    }


    final class MysqlEventListener implements BinaryLogRemoteClient.EventListener {

        @Override
        public void onEvent(Object sender, Event event) {
            // ROTATE > FORMAT_DESCRIPTION > TABLE_MAP > WRITE_ROWS > UPDATE_ROWS > DELETE_ROWS > XID
            BinaryLogClient binaryLogClient = (BinaryLogClient)sender;
            EventHeader header = event.getHeader();
            if (header.getEventType() == EventType.XID) {
                refresh(header);
                return;
            }
            // 切换binlog
            if (header.getEventType() == EventType.ROTATE) {
                RotateEventData data = event.getData();
                refresh(data.getBinlogFilename(), data.getBinlogPosition());
                return;
            }


            if (EventType.isUpdate(header.getEventType())) {
                refresh(header);
                UpdateRowsEventData data = event.getData();
                DBTableInfo dbTableInfo = procDataChange(binaryLogClient,data.getTableId());
                if(filterTable(dbTableInfo)) {
                    data.getRows().forEach(item -> {
                        Map<String, Object> dataMap = createMap(dbTableInfo, item.getValue());
                        listeners.forEach(listener ->
                                listener.onEvents(builderRowChangedEvent(dbTableInfo, ConnectorConstant.OPERTION_UPDATE, dataMap)));
                    });
                }
                return;
            }
            if (EventType.isWrite(header.getEventType())) {
                refresh(header);
                WriteRowsEventData data = event.getData();
                DBTableInfo dbTableInfo = procDataChange(binaryLogClient,data.getTableId());
                if(filterTable(dbTableInfo)) {
                    data.getRows().forEach(item -> {
                        Map<String, Object> dataMap = createMap(dbTableInfo, item);
                        listeners.forEach(listener ->
                                listener.onEvents(builderRowChangedEvent(dbTableInfo, ConnectorConstant.OPERTION_INSERT, dataMap)));
                    });
                }
                return;
            }
            if (EventType.isDelete(header.getEventType())) {
                refresh(header);
                DeleteRowsEventData data = event.getData();
                DBTableInfo dbTableInfo = procDataChange(binaryLogClient,data.getTableId());
                if(filterTable(dbTableInfo)) {
                    data.getRows().forEach(item -> {
                        Map<String, Object> dataMap = createMap(dbTableInfo, item);
                        listeners.forEach(listener ->
                                listener.onEvents(builderRowChangedEvent(dbTableInfo, ConnectorConstant.OPERTION_DELETE, dataMap)));
                    });
                }
                return;
            }

        }


        private DBTableInfo procDataChange(BinaryLogClient binaryLogClient, long tableId){

            TableMapEventData tableMapEventData = binaryLogClient.getTable(tableId);

            if(dataBaseAccess.getDataSource().getDatabaseName().equals(tableMapEventData.getDatabase())) {
                DBTableInfo dbTableInfo = tables.get(tableMapEventData.getTable());
                if (dbTableInfo != null) {
                    dbTableInfo.setTableID(tableId + "");
                }
                return dbTableInfo;
            }
            return null;
        }
    }

}
